/* ----------------------------------------------------------------------------
 * Copyright (c) Huawei Technologies Co., Ltd. 2013-2020. All rights reserved.
 * Description: Mqtt Client
 * Author: Huawei LiteOS Team
 * Create: 2013-01-01
 * Redistribution and use in source and binary forms, with or without modification,
 * are permitted provided that the following conditions are met:
 * 1. Redistributions of source code must retain the above copyright notice, this list of
 * conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright notice, this list
 * of conditions and the following disclaimer in the documentation and/or other materials
 * provided with the distribution.
 * 3. Neither the name of the copyright holder nor the names of its contributors may be used
 * to endorse or promote products derived from this software without specific prior written
 * permission.
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO,
 * THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
 * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR
 * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
 * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
 * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS;
 * OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
 * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR
 * OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
 * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 * --------------------------------------------------------------------------- */

#include "mqtt_client.h"
#include "los_base.h"
#include "los_task_pri.h"
#include "los_typedef.h"
#include "los_sys.h"
#include "MQTTClient.h"
#include "flash_manager.h"
#include "dtls_interface.h"
#include "cJSON.h"
#include "hmac.h"
#include "atiny_log.h"

#ifdef __cplusplus
#if __cplusplus
extern "C" {
#endif /* __cplusplus */
#endif /* __cplusplus */

#define MQTT_VERSION_3_1 (3)
#define MQTT_VERSION_3_1_1 (4)

#define VARIABLE_SIZE (2)
#define TOPIC_ROOT_FMT "$oc/devices/%s/sys/#" // mqtt root topic

#define IS_VALID_NAME_LEN(name) (strnlen((name), MQTT_STRING_MAX_LEN + 1) <= MQTT_STRING_MAX_LEN)

typedef enum {
    MQTT_CONNECT_WITH_PRODUCT_ID,
    MQTT_CONNECT_WITH_DEVICE_ID
} mqtt_dynamic_connect_state_e;

typedef struct {
    mqtt_static_connection_info_s save_info;
    char *get_secret;
    mqtt_dynamic_connect_state_e state;
    uint8_t connection_update_flag;
    uint8_t has_device_id;
    uint8_t reserve[2];
} mqtt_dynamic_info_s;

struct mqtt_client_tag_s {
    mqtt_device_info_s device_info;
    MQTTClient client;
    mqtt_param_s params;
    mqtt_dynamic_info_s dynamic_info;
    char *sub_topic;
    uint8_t init_flag;
    uint8_t reserve[3];
};

static uint8_t g_mqtt_sendbuf[MQTT_SENDBUF_SIZE];

/* reserve 1 byte for string end 0 for jason */
static uint8_t g_mqtt_readbuf[MQTT_READBUF_SIZE + 1];

static mqtt_client_s g_mqtt_client;

static int mqtt_cmd_ioctl(mqtt_cmd_e cmd, void *arg, int32_t len)
{
    mqtt_client_s *handle = &g_mqtt_client;
    if (handle->params.cmd_ioctl != NULL) {
        return handle->params.cmd_ioctl(cmd, arg, len);
    }
    ATINY_LOG(LOG_FATAL, "cmd_ioctl null");
    return ATINY_ERR;
}

static void mqtt_free_params(mqtt_param_s *param)
{
    TRY_FREE_MEM(param->server_ip);
    TRY_FREE_MEM(param->server_port);
    switch (param->info.security_type) {
        case MQTT_SECURITY_TYPE_PSK:
            TRY_FREE_MEM(param->info.u.psk.psk_id);
            TRY_FREE_MEM(param->info.u.psk.psk);
            break;
        case MQTT_SECURITY_TYPE_CA:
            TRY_FREE_MEM(param->info.u.ca.ca_crt);
            break;
        default:
            break;
    }
}

static int mqtt_check_param(const mqtt_param_s *param)
{
    if ((param->server_ip == NULL)
        || (param->server_port == NULL)
        || (param->info.security_type >= MQTT_SECURITY_TYPE_MAX)
        || (param->cmd_ioctl == NULL)) {
        ATINY_LOG(LOG_FATAL, "invalid param, sec type %d", param->info.security_type);
        return ATINY_ARG_INVALID;
    }
    if (param->info.security_type == MQTT_SECURITY_TYPE_PSK) {
        if ((param->info.u.psk.psk == NULL)
            || (param->info.u.psk.psk_len <= 0)
            || (param->info.u.psk.psk_id == NULL)
            || (param->info.u.psk.psk_id_len <= 0)) {
            ATINY_LOG(LOG_FATAL, "invalid psk");
            return ATINY_ARG_INVALID;
        }
    }
    if (param->info.security_type == MQTT_SECURITY_TYPE_CA) {
        if (param->info.u.ca.ca_crt == NULL) {
            ATINY_LOG(LOG_FATAL, "invalid ca");
            return ATINY_ARG_INVALID;
        }
    }
    return ATINY_OK;
}

static int mqtt_dup_param(mqtt_param_s *dest, const mqtt_param_s *src)
{
    memset(dest, 0, sizeof(*dest));

    dest->info.security_type = src->info.security_type;
    dest->cmd_ioctl = src->cmd_ioctl;

    dest->server_ip = atiny_strdup(src->server_ip);
    if (dest->server_ip == NULL) {
        ATINY_LOG(LOG_FATAL, "atiny_strdup NULL");
        return ATINY_MALLOC_FAILED;
    }

    dest->server_port = atiny_strdup(src->server_port);
    if (dest->server_port == NULL) {
        ATINY_LOG(LOG_FATAL, "atiny_strdup NULL");
        goto mqtt_param_dup_failed;
    }

    switch (src->info.security_type) {
        case MQTT_SECURITY_TYPE_PSK:
            dest->info.u.psk.psk_id = (uint8_t *)atiny_malloc(src->info.u.psk.psk_id_len);
            if (dest->info.u.psk.psk_id == NULL) {
                ATINY_LOG(LOG_FATAL, "atiny_strdup NULL");
                goto mqtt_param_dup_failed;
            }
            memset(dest->info.u.psk.psk_id, 0, src->info.u.psk.psk_id_len);
            memcpy(dest->info.u.psk.psk_id, src->info.u.psk.psk_id, src->info.u.psk.psk_id_len);
            dest->info.u.psk.psk_id_len = src->info.u.psk.psk_id_len;
            dest->info.u.psk.psk_id[dest->info.u.psk.psk_id_len] = '\0';
            dest->info.u.psk.psk = (unsigned char *)atiny_malloc(src->info.u.psk.psk_len);
            if (dest->info.u.psk.psk == NULL) {
                ATINY_LOG(LOG_FATAL, "atiny_strdup NULL");
                goto mqtt_param_dup_failed;
            }
            memset(dest->info.u.psk.psk, 0, src->info.u.psk.psk_len);
            memcpy(dest->info.u.psk.psk, src->info.u.psk.psk, src->info.u.psk.psk_len);
            dest->info.u.psk.psk_len = src->info.u.psk.psk_len;
            break;
        case MQTT_SECURITY_TYPE_CA:
            dest->info.u.ca.ca_crt = (char *)atiny_malloc(src->info.u.ca.ca_len);
            if (dest->info.u.ca.ca_crt == NULL) {
                ATINY_LOG(LOG_FATAL, "atiny_strdup NULL");
                goto mqtt_param_dup_failed;
            }
            memcpy(dest->info.u.ca.ca_crt, src->info.u.ca.ca_crt, src->info.u.ca.ca_len);
            dest->info.u.ca.ca_len = src->info.u.ca.ca_len;
            break;
        default:
            break;
    }
    return ATINY_OK;
mqtt_param_dup_failed:
    mqtt_free_params(dest);
    return ATINY_MALLOC_FAILED;
}

static void mqtt_free_device_info(mqtt_device_info_s *info)
{
    TRY_FREE_MEM(info->secret);
    if (MQTT_STATIC_CONNECT == info->connection_type) {
        TRY_FREE_MEM(info->u.s_info.deviceid);
    } else {
        TRY_FREE_MEM(info->u.d_info.productid);
        TRY_FREE_MEM(info->u.d_info.nodeid);
    }
}

static int mqtt_check_device_info(const mqtt_device_info_s *info)
{
    if ((info->connection_type >= MQTT_MAX_CONNECTION_TYPE)
        || (info->codec_mode >= MQTT_MAX_CODEC_MODE)
        || (info->sign_type >= MQTT_MAX_SIGN_TYPE)
        || (info->secret == NULL)
        || (!IS_VALID_NAME_LEN(info->secret))) {
        ATINY_LOG(LOG_FATAL, "invalid device info con_type %d codec_mode %d signe type %d",
            info->connection_type, info->codec_mode, info->sign_type);
        return ATINY_ARG_INVALID;
    }

    if ((info->connection_type == MQTT_STATIC_CONNECT)
        && ((info->u.s_info.deviceid == NULL)
        || (!IS_VALID_NAME_LEN(info->u.s_info.deviceid)))) {
        ATINY_LOG(LOG_FATAL, "invalid static device info con_type %d codec_mode %d signe type %d",
            info->connection_type, info->codec_mode, info->sign_type);
        return ATINY_ARG_INVALID;
    }

    if ((info->connection_type == MQTT_DYNAMIC_CONNECT)
        &&((info->u.d_info.productid == NULL)
        || (!IS_VALID_NAME_LEN(info->u.d_info.productid))
        || (info->u.d_info.nodeid == NULL)
        || !(IS_VALID_NAME_LEN(info->u.d_info.nodeid)))) {
        ATINY_LOG(LOG_FATAL, "invalid dynamic device info con_type %d codec_mode %d signe type %d",
            info->connection_type, info->codec_mode, info->sign_type);
        return ATINY_ARG_INVALID;
    }
    return ATINY_OK;
}

static int mqtt_dup_device_info(mqtt_device_info_s *dest, const mqtt_device_info_s *src)
{
    memset(dest, 0, sizeof(*dest));
    dest->connection_type = src->connection_type;
    dest->sign_type = src->sign_type;
    dest->codec_mode = src->codec_mode;
    dest->secret = atiny_strdup(src->secret);
    if (dest->secret == NULL) {
        ATINY_LOG(LOG_FATAL, "atiny_strdup fail");
        return ATINY_MALLOC_FAILED;
    }

    if (MQTT_STATIC_CONNECT == src->connection_type) {
        dest->u.s_info.deviceid = atiny_strdup(src->u.s_info.deviceid);
        if (dest->u.s_info.deviceid == NULL) {
            ATINY_LOG(LOG_FATAL, "atiny_strdup fail");
            goto MALLOC_FAIL;
        }
    } else {
        dest->u.d_info.productid = atiny_strdup(src->u.d_info.productid);
        if (dest->u.d_info.productid == NULL) {
            ATINY_LOG(LOG_FATAL, "atiny_strdup fail");
            goto MALLOC_FAIL;
        }
        dest->u.d_info.nodeid = atiny_strdup(src->u.d_info.nodeid);
        if (dest->u.d_info.nodeid == NULL) {
            ATINY_LOG(LOG_FATAL, "atiny_strdup fail");
            goto MALLOC_FAIL;
        }
    }
    return ATINY_OK;
MALLOC_FAIL:
    mqtt_free_device_info(dest);
    return ATINY_MALLOC_FAILED;
}

static bool mqtt_is_connectting_with_deviceid(const mqtt_client_s *handle)
{
    return (handle->device_info.connection_type == MQTT_STATIC_CONNECT);
}

static char *mqtt_add_strings(const char *strs[], uint32_t tmp_len[], uint32_t str_num)
{
    uint32_t i;
    const char hyphen = '_';
    uint32_t len = 0;
    char *result;
    char *cur;

    for (i = 0; i < str_num; i++) {
        tmp_len[i] = strnlen(strs[i], STRING_MAX_LEN);
        len += (tmp_len[i] + 1);
    }

    result = atiny_malloc(len);
    if (result == NULL) {
        ATINY_LOG(LOG_FATAL, "mqtt_cmd_ioctl fail");
        return NULL;
    }

    cur = result;
    for (i = 0; i < str_num; i++) {
        memcpy(cur, strs[i], tmp_len[i]);
        cur += tmp_len[i];
        if (i != str_num - 1) {
            *cur = hyphen;
            cur++;
        }
    }
    *cur = '\0';
    return result;
}

static const char *mqtt_connection_type_to_str(mqtt_connection_type_e type)
{
    return MQTT_STATIC_CONNECT == type ? "0" : "1";
}

static const char *mqtt_sign_type_to_str(mqtt_password_sign_type_e type)
{
    return MQTT_SIGN_TYPE_HMACSHA256_NO_CHECK_TIME == type ? "0" : "1";
}

static void mqtt_bin_to_str(const uint8_t *bin_buf, char *str_buf, uint32_t bin_len)
{
    uint32_t i;

    for (i = 0; i < bin_len; i++) {
        (void)snprintf(str_buf + i * 2, 3, "%02x", bin_buf[i]);
    }
}
static char *mqtt_get_send_password(char *pwd, char *time)
{
    uint8_t hmac[32];
    const uint32_t len = sizeof(hmac) * 2 + 1;
    char *result;
    int ret;
    mbedtls_hmac_t hmac_info;

    hmac_info.secret = (uint8_t *)time;
    hmac_info.secret_len = strnlen(time, MQTT_TIME_BUF_LEN);
    hmac_info.input = (uint8_t *)pwd;
    hmac_info.input_len = strnlen(pwd, STRING_MAX_LEN);
    hmac_info.digest = hmac;
    hmac_info.digest_len = sizeof(hmac);
    hmac_info.hmac_type = MBEDTLS_MD_SHA256;
    ret = mbedtls_hmac_calc(&hmac_info);
    if (ret != ATINY_OK) {
        ATINY_LOG(LOG_FATAL, "mbedtls_hmac_calc fail,ret %d", ret);
        return NULL;
    }

    result = atiny_malloc(len);
    if (result == NULL) {
        ATINY_LOG(LOG_FATAL, "atiny_malloc fail");
        return NULL;
    }
    mqtt_bin_to_str(hmac, result, sizeof(hmac));
    return result;
}

static void mqtt_destroy_data_connection_info(MQTTPacket_connectData *data)
{
    TRY_FREE_MEM(data->clientID.cstring);
    TRY_FREE_MEM(data->password.cstring);
}

static int mqtt_get_connection_info(mqtt_client_s *handle, MQTTPacket_connectData *data)
{
    char *strs[5];
    uint32_t tmp[array_size(strs)];
    uint32_t str_num = 0;
    char time[MQTT_TIME_BUF_LEN];
    char *pwd = NULL;

    if (mqtt_is_connectting_with_deviceid(handle)) {
        if (handle->device_info.connection_type == MQTT_STATIC_CONNECT) {
            strs[0] = handle->device_info.u.s_info.deviceid;
            pwd = handle->device_info.secret;
        }
        str_num = 1;
        ATINY_LOG(LOG_INFO, "try static connect");
    } else {
        strs[0] = handle->device_info.u.d_info.productid;
        strs[1] = handle->device_info.u.d_info.nodeid;
        str_num = 2;
        pwd = handle->device_info.secret;
        ATINY_LOG(LOG_INFO, "try dynamic connect");
    }

    strs[str_num++] = (char *)mqtt_connection_type_to_str(
            mqtt_is_connectting_with_deviceid(handle) ? MQTT_STATIC_CONNECT : MQTT_DYNAMIC_CONNECT);
    strs[str_num++] = (char *)mqtt_sign_type_to_str(handle->device_info.sign_type);
    if (mqtt_cmd_ioctl(MQTT_GET_TIME, time, sizeof(time)) != ATINY_OK) {
        ATINY_LOG(LOG_FATAL, "mqtt_cmd_ioctl fail");
        return ATINY_ERR;
    }
    time[sizeof(time) - 1] = '\0';
    strs[str_num++] = time;

    data->clientID.cstring = mqtt_add_strings((const char **)strs, tmp, str_num);
    if (data->clientID.cstring == NULL) {
        return ATINY_MALLOC_FAILED;
    }

    data->username.cstring = strs[0]; /* deviceid or pruoductid */
    data->password.cstring = mqtt_get_send_password(pwd, time);
    if (data->password.cstring == NULL) {
        return ATINY_ERR;
    }

    ATINY_LOG(LOG_DEBUG, "send user %s client %s", data->username.cstring, data->clientID.cstring);
    return ATINY_OK;
}

char *mqtt_get_topic(const char *fmt, uint32_t fixed_size)
{
    uint32_t len;
    char *deviceId = NULL;
    mqtt_client_s *handle = &g_mqtt_client;
    if (handle->device_info.connection_type == MQTT_STATIC_CONNECT) {
        deviceId = handle->device_info.u.s_info.deviceid;
    }

    len = fixed_size + strnlen(deviceId, MQTT_STRING_MAX_LEN);
    if (len < 0) {
        return NULL;
    }
    char *topic = atiny_malloc(len);
    if (topic == NULL) {
        ATINY_LOG(LOG_FATAL, "atiny_malloc failed, len %d", len);
        return NULL;
    }
    (void)snprintf(topic, len, fmt, deviceId);
    return topic;
}

static void mqtt_subscribe_callback(MessageData *md)
{
    if (md == NULL) {
        return;
    }
    if (md->message == NULL || md->topicName == NULL) {
        return;
    }
    /* add for json parse,then not need to copy in callback */
    if (md->message->payloadlen > sizeof(g_mqtt_readbuf)) {
        return;
    }
    char *topic = md->topicName->lenstring.data;
    int topicLen = md->topicName->lenstring.len;
	MQTTLenString *lenstring = (MQTTLenString *)&md->topicName->lenstring;
    (void)mqtt_cmd_ioctl(MQTT_RCV_MSG, md, sizeof(md));
}

static int mqtt_subscribe_topic(mqtt_client_s *handle, char *topic, messageHandler callback)
{
    int rc = FAILURE;
    if (handle == NULL || callback == NULL || topic == NULL) {
        return rc;
    }
    if (handle->sub_topic != NULL) {
        (void)MQTTSetMessageHandler(&handle->client, handle->sub_topic, NULL);
        atiny_free(handle->sub_topic);
        handle->sub_topic = NULL;
    }

    rc = MQTTSubscribe(&handle->client, topic, QOS1, callback);
    if (rc == MQTT_SUCCESS) {
        handle->sub_topic = topic;
    } else {
        ATINY_LOG(LOG_FATAL, "MQTTSubscribe failed, rc=%d, topic=%s", rc, topic);
    }
    return rc;
}

static void mqtt_disconnect( MQTTClient *client, Network *n)
{
    if (MQTTIsConnected(client)) {
        (void)MQTTDisconnect(client);
    }
    NetworkDisconnect(n);
    ATINY_LOG(LOG_ERR, "mqtt_disconnect");
}

static mqtt_security_info_s *mqtt_get_security_info(void *param)
{
    (void)param;
    mqtt_client_s *handle = &g_mqtt_client;
    return &handle->params.info;
}

int atiny_mqtt_init(const mqtt_param_s *params)
{
    mqtt_client_s *phandle = &g_mqtt_client;
    if ((params == NULL) || (phandle == NULL)
        || (params->info.security_type > MQTT_SECURITY_TYPE_CA)
        || (mqtt_check_param(params) != ATINY_OK)) {
        ATINY_LOG(LOG_FATAL, "Invalid args");
        return ATINY_ARG_INVALID;
    }
    if (phandle->init_flag) {
        ATINY_LOG(LOG_FATAL, "mqtt reinit");
        return ATINY_ERR;
    }
    memset(phandle, 0, sizeof(mqtt_client_s));
    if (ATINY_OK != mqtt_dup_param(&(phandle->params), params)) {
        return ATINY_MALLOC_FAILED;
    }
    flash_manager_init(mqtt_cmd_ioctl);
    phandle->init_flag = true;
    return ATINY_OK;
}

static int atiny_mqtt_bind_quit(mqtt_client_s *handle)
{
    int result = ATINY_ERR;
    mqtt_free_params(&(handle->params));
    (void)atiny_task_mutex_lock(&(handle->client.mutex));
    mqtt_free_device_info(&(handle->device_info));
    (void)atiny_task_mutex_unlock(&(handle->client.mutex));
    MQTTClientDeInit(&handle->client);
    handle->init_flag = false;
    return result;
}

int atiny_mqtt_bind(const mqtt_device_info_s *device_info)
{
    Network n;
    int rc;
    MQTTClient *client = NULL;
    mqtt_param_s *params = NULL;
    MQTTPacket_connectData data = MQTTPacket_connectData_initializer;
    Timer timer;
    int result = ATINY_ERR;
    bool connectRetry = 0;
    char *topic = NULL;

    if ((device_info == NULL) || (mqtt_check_device_info(device_info) != ATINY_OK)) {
        ATINY_LOG(LOG_FATAL, "parameter invalid");
        result = ATINY_ARG_INVALID;
        atiny_mqtt_bind_quit(&g_mqtt_client);
        return result;
    }

    dtls_init();

    client = &(g_mqtt_client.client);
    params = &(g_mqtt_client.params);

    rc = mqtt_dup_device_info(&(g_mqtt_client.device_info), device_info);
    if (rc != ATINY_OK) {
        atiny_mqtt_bind_quit(&g_mqtt_client);
        return result;
    }

    (void)memset_s(&n, sizeof(Network), 0x0, sizeof(Network));
    NetworkInit(&n, mqtt_get_security_info);

    memset(client, 0x0, sizeof(MQTTClient));
    rc = MQTTClientInit(client, &n, MQTT_COMMAND_TIMEOUT_MS, g_mqtt_sendbuf, MQTT_SENDBUF_SIZE, g_mqtt_readbuf, MQTT_READBUF_SIZE);
    if (rc != MQTT_SUCCESS) {
        ATINY_LOG(LOG_FATAL, "MQTTClientInit fail,rc %d", rc);
        atiny_mqtt_bind_quit(&g_mqtt_client);
        return result;
    }

    data.willFlag = 0;
    data.MQTTVersion = MQTT_VERSION_3_1_1;
    data.keepAliveInterval = MQTT_KEEPALIVE_INTERVAL_S;
    data.cleansession = true;

    while (1) {
        if (connectRetry) {
            ATINY_LOG(LOG_INFO, "reconnect delay[%d]", MQTT_CONN_RETRY_DELAY);
            (void)LOS_TaskDelay(MQTT_CONN_RETRY_DELAY);
        }
        rc = NetworkConnect(&n, params->server_ip, atoi(params->server_port));
        if (rc != 0) {
            ATINY_LOG(LOG_ERR, "NetworkConnect failed: %d", rc);
            connectRetry = true;
            continue;
        }

        if (mqtt_get_connection_info(&g_mqtt_client, &data) != ATINY_OK) {
            mqtt_destroy_data_connection_info(&data);
            connectRetry = true;
            mqtt_disconnect(client, &n);
            continue;
        }

        rc = MQTTConnect(client, &data);
        mqtt_destroy_data_connection_info(&data);
        ATINY_LOG(LOG_DEBUG, "CONNACK : %d.", rc);
        if (rc != MQTT_SUCCESS) {
            ATINY_LOG(LOG_ERR, "MQTTConnect failed, rc=%d.", rc);
            continue;
        }

        if (mqtt_is_connectting_with_deviceid(&g_mqtt_client)) {
            topic = mqtt_get_topic(TOPIC_ROOT_FMT, sizeof(TOPIC_ROOT_FMT) - VARIABLE_SIZE);
            if (topic == NULL) {
                ATINY_LOG(LOG_ERR, "topic is null. ");
                connectRetry = true;
                continue;
            }
            ATINY_LOG(LOG_INFO, "try subcribe static topic=%s", topic);
        }
        if (ATINY_OK != mqtt_subscribe_topic(&g_mqtt_client, topic, mqtt_subscribe_callback)) {
            ATINY_LOG(LOG_ERR, "mqtt_subscribe_topic failed.");
            connectRetry = true;
            continue;
        }

        connectRetry = false;
        if (!mqtt_is_connectting_with_deviceid(&g_mqtt_client)) {
            TimerInit(&timer);
            TimerCountdownMS(&timer, MQTT_WRITE_FOR_SECRET_TIMEOUT);
        }
        while (rc >= 0 && MQTTIsConnected(client)) {
            rc = MQTTYield(client, MQTT_EVENTS_HANDLE_PERIOD_MS);
        }
        mqtt_disconnect(client, &n);
    }
    (void)atiny_mqtt_bind_quit(&g_mqtt_client);
    return ATINY_OK;
}

int atiny_mqtt_data_send(char *topic, const char *msg, uint32_t msg_len, mqtt_qos_e qos)
{
    MQTTMessage message;
    int rc;
    size_t payloadLen;
    mqtt_client_s *phandle = &g_mqtt_client;
    if ((phandle == NULL) || (qos >= MQTT_QOS_MAX)) {
        ATINY_LOG(LOG_FATAL, "parameter invalid");
        return ATINY_ARG_INVALID;
    }
    if (topic == NULL) {
        ATINY_LOG(LOG_FATAL, "topic invalid");
        return ATINY_ARG_INVALID;
    }
    if (msg == NULL || msg_len <= 0) {
        ATINY_LOG(LOG_FATAL, "msg invalid");
        return ATINY_ARG_INVALID;
    }
    if (phandle->device_info.codec_mode == MQTT_CODEC_MODE_JSON) {
        payloadLen = strnlen(msg, msg_len);
    } else {
        payloadLen = msg_len;
    }

    if (!atiny_mqtt_isconnected(phandle)) {
        ATINY_LOG(LOG_FATAL, "not connected");
        return ATINY_ERR;
    }

    memset(&message, 0, sizeof(message));
    message.qos = (enum QoS)qos;
    message.payload = (void *)msg;
    message.payloadlen = payloadLen;
    rc = MQTTPublish(&phandle->client, topic, &message);
    if (rc != MQTT_SUCCESS) {
        ATINY_LOG(LOG_FATAL, "MQTTPublish failed, rc %d", rc);
        return ATINY_ERR;
    }
    return ATINY_OK;
}

int atiny_mqtt_isconnected(mqtt_client_s *phandle)
{
    if (phandle == NULL) {
        ATINY_LOG(LOG_ERR, "invalid args");
        return false;
    }
    return mqtt_is_connectting_with_deviceid(phandle) && MQTTIsConnected(&(phandle->client));
}

char *atiny_mqtt_get_deviceId(mqtt_client_s *phandle)
{
    if (phandle->device_info.connection_type == MQTT_STATIC_CONNECT) {
        return phandle->device_info.u.s_info.deviceid;
    } else {
        return phandle->device_info.u.d_info.productid;
    }
}

#ifdef __cplusplus
#if __cplusplus
}
#endif /* __cplusplus */
#endif /* __cplusplus */
